package dai.samples.batch.multi;

import dai.samples.batch.entity.BatchEntity;
import dai.samples.batch.config.DbWriter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.*;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
import org.springframework.batch.item.json.JacksonJsonObjectReader;
import org.springframework.batch.item.json.builder.JsonFileItemWriterBuilder;
import org.springframework.batch.item.json.builder.JsonItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.UUID;

/**
 * TODO 未完成
 * @author daify
 * @date 2020-11-15
 */
@Slf4j
@Configuration
public class MultiJobConfig {


    @Autowired
    JobBuilderFactory jobBuilderFactory;

    @Autowired
    StepBuilderFactory stepBuilderFactory;

    @Autowired
    PlatformTransactionManager transactionManager;

    /**
     * 数据写入数据库
     */
    @Autowired
    DbWriter dbWriter;

    @Autowired
    DataSource dataSource;


    /**
     * 任务
     * @param jobRepository
     * @return
     */
    @Bean("multiJob")
    public Job multiJob(JobRepository jobRepository) {
        return this.jobBuilderFactory.get("multiJob")
                .repository(jobRepository)
                .start(readStep())
                .next(multiStep())
                .build();
    }


    /**
     * 任务
     * @param jobRepository
     * @return
     */
    @Bean("multiJob2")
    public Job multiJob2(JobRepository jobRepository) {
        return this.jobBuilderFactory.get("multiJob")
                .repository(jobRepository)
                .start(readStep())
                .next(multiStep2())
                .build();
    }


    @Bean
    public TaskExecutor taskExecutor(){
        return new SimpleAsyncTaskExecutor("spring_batch");
    }

    @Bean(value = "readStep")
    public Step readStep() {
        return this.stepBuilderFactory.get("readStep")
                .<BatchEntity, BatchEntity>chunk(2)
                .reader(itemReader())
                .processor(new AddIdProcessor())
                .writer(dbWriter)
                .build();
    }

    /**
     * 使用了多线程块处理
     * @return
     */
    @Bean(value = "multiStep2")
    public Step multiStep2() {
        return this.stepBuilderFactory.get("multiStep")
                .<BatchEntity, BatchEntity>chunk(2)
                .reader(getDbRead())
                .processor(getProcessor())
                .writer(itemWriter())
                .taskExecutor(taskExecutor())
                .throttleLimit(4)
                .build();
    }

    /**
     * 正常的块处理
     * @return
     */
    @Bean(value = "multiStep")
    public Step multiStep() {
        return this.stepBuilderFactory.get("multiStep")
                //.transactionManager(transactionManager)
                .<BatchEntity, BatchEntity>chunk(2)
                .reader(getDbRead())
                .processor(getProcessor())
                .writer(itemWriter())
                /*.taskExecutor(taskExecutor())
                .throttleLimit(4)*/
                .build();
    }

    /**
     * JSON写数据
     * @return
     */
    private ItemWriter<BatchEntity> itemWriter() {
        String name = "multiJob-" + System.currentTimeMillis();
        String patch = "target/test-outputs/" + name + ".json";
        return new JsonFileItemWriterBuilder<BatchEntity>()
                .jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>())
                .resource(new FileSystemResource(patch))
                .name(name)
                .build();
    }


    /**
     * JSON读数据
     * @return
     */
    public ItemReader<BatchEntity> itemReader() {
        return new JsonItemReaderBuilder<BatchEntity>()
                .jsonObjectReader(new JacksonJsonObjectReader<>(BatchEntity.class))
                .resource(new ClassPathResource("data/batchJob.json"))
                .name("batchJobReader")
                .strict(false)
                .build();
    }

    public ItemProcessor<? super BatchEntity, ? extends BatchEntity> getProcessor() {
        return new ChangeNameProcessor();
    }

    /**
     * 从DB中读取数据
     * @return
     */
    public ItemReader<BatchEntity> getDbRead() {
        JdbcCursorItemReader itemReader = new JdbcCursorItemReader();
        itemReader.setDataSource(dataSource);
        itemReader.setSql("SELECT entity_id,age,first_name,full_name,hello_message,is_adult,last_name FROM batch_entity order by age");
        itemReader.setRowMapper(rowMapper());
        return itemReader;
    }

    private RowMapper<BatchEntity> rowMapper() {
        return new BeanPropertyRowMapper<>(BatchEntity.class);
    }


    /**
     * 处理过程 打印数据库中的数据
     */
    class ChangeNameProcessor implements ItemProcessor<BatchEntity, BatchEntity> {

        @Override
        public BatchEntity process(BatchEntity person) {
            String fullName = person.getFullName();
            System.out.println(fullName + ":" + person.getAge());
            return person;
        }
    }


    /**
     * 处理过程，设置ID
     */
    class AddIdProcessor implements ItemProcessor<BatchEntity, BatchEntity> {

        @Override
        public BatchEntity process(BatchEntity person) {
            String id = UUID.randomUUID().toString();
            person.setEntityId(id);
            String fullName = person.getFullName();
            person.setFullName(fullName);
            return person;
        }
    }
}
